package com.azh.server.datam.dataItem;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.azh.server.common.SubArray;
import com.azh.server.datam.DataManagerImpl;
import com.azh.server.datam.page.Page;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-25
 * DataItem数据格式: [validFlag][DataSize][Data]
 *
 * 修改数据流程：
 *  修改前调用before方法，修改后调用unBefore方法，修改完成后调用after方法
 *  整个流程是为了保存前向数据，同时及时落日志，DataManager保证DataItem修改是原子性的
 */
public class DataItemImpl implements DataItem{

    public static final int of_valid = 0;
    public static final int of_size = 1;
    public static final int of_data = 3;

    private SubArray raw;
    private byte[] oldRaw;
    // dm 主要是因为dataItem的释放依赖dm的释放，
    private DataManagerImpl dm;
    private long uid;
    private Page page;
    private Lock rLock;
    private Lock wLock;

    public DataItemImpl(SubArray raw, byte[] oldRaw, Page page, long uid, DataManagerImpl dmi) {
        this.raw = raw;
        this.oldRaw = oldRaw;
        ReadWriteLock lock = new ReentrantReadWriteLock();
        rLock = lock.readLock();
        wLock = lock.writeLock();
        this.dm = dmi;
        this.uid = uid;
        this.page = page;
    }

    public boolean isValid() {
        return raw.raw[raw.start + of_valid] == (byte) 0;
    }

    /**
     * 上层模块获取到DataItem后，可以通过该方法获取数据，SubArray是共享数据
     * @return
     */
    @Override
    public SubArray data() {
        return new SubArray(raw.raw, raw.start + of_data, raw.end);
    }

    @Override
    public void before() {
        wLock.lock();
        page.setDirty(true);
        System.arraycopy(raw.raw, raw.start, oldRaw, 0, oldRaw.length);
    }

    @Override
    public void unBefore() {
        System.arraycopy(oldRaw, 0, raw.raw, raw.start, oldRaw.length);
        wLock.unlock();
    }

    @Override
    public void after(long txId) {
        dm.logDataItem(txId, this);
        wLock.unlock();
    }

    @Override
    public void release() {
        dm.releaseDataItem(this);
    }

    @Override
    public void lock() {
        wLock.lock();
    }

    @Override
    public void unlock() {
        wLock.unlock();
    }

    @Override
    public void rLock() {
        rLock.lock();
    }

    @Override
    public void rUnLock() {
        rLock.unlock();
    }

    @Override
    public Page page() {
        return page;
    }

    @Override
    public long getUid() {
        return uid;
    }

    @Override
    public byte[] getOldRaw() {
        return oldRaw;
    }

    @Override
    public SubArray getRaw() {
        return raw;
    }
}
